package com.bstek.sample.redisSample;

import cn.hutool.core.bean.BeanUtil;
import com.bstek.bdf2.core.redis.BstekRedisCache;
import com.bstek.sample.domain.Order;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @Title: 基于redis队列和线程的方式轮询消费订单
 * @Package: com.bstek.bdf2.core.cache
 * @Description:
 * @Author: zxq
 * @Date: 2024/1/24 - 16:55
 */
@Slf4j
@Service
public class QueueRedisService {

    @Resource
    private BstekRedisCache bstekRedisCache;

    private final StringRedisTemplate stringRedisTemplate;

    public QueueRedisService(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate=stringRedisTemplate;
    }


    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

//    @PostConstruct
    private void init() {
        SECKILL_ORDER_EXECUTOR.submit(new TaskHandler());
    }


    private class TaskHandler implements Runnable{

        @Override
        public void run() {
            while (true){
                try {
                    // 1.获取消息队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
                    );
                    // 2.判断订单信息是否为空
                    if (list == null || list.isEmpty()) {
                        // 如果为null，说明没有消息，继续下一次循环
                        continue;
                    }
                    // 解析数据
                    buildData(list);

                }catch (Exception e){
                    log.info(e.getMessage());
                    handleExceptionList();
                }
            }
        }
    }

    private void buildData(List<MapRecord<String, Object, Object>> list) {
        MapRecord<String, Object, Object> record = list.get(0);
        Map<Object, Object> value = record.getValue();
        Order order = BeanUtil.fillBeanWithMap(value, new Order(), true);
        // 3.创建订单
        createOrder(order);
        // 4.确认消息 XACK
        stringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId());
    }

    private void handleExceptionList() {
        while (true) {
            try {
                // 1.获取pending-list中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 0
                List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1),
                        StreamOffset.create("stream.orders", ReadOffset.from("0"))
                );
                // 2.判断订单信息是否为空
                if (list == null || list.isEmpty()) {
                    // 如果为null，说明没有异常消息，结束循环
                    break;
                }
                // 解析数据
                buildData(list);
            } catch (Exception e) {
                log.error("处理订单异常", e);
            }
        }
    }


    private void createOrder(Order order) {
        long nextId = bstekRedisCache.nextId("order");

    }
}
